package org.shj.spark.mlib.videolearn

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import java.text.SimpleDateFormat
import java.util.Date
import org.apache.spark.sql.SparkSession

/**
 * 关联规则用于表示数据内隐含的关联性，例如表示购买了尿布的消费者往往也会购买啤酒。
 * 一般记 X 为先决条件，Y 为相应的关联结果
 * 关联性强度如何，由3个概念： 支持度，置信度和提升度来控制和评价
 * 例如： 有10000 个消费者购买了商品，其中购买尿布的有 1000 人，购买啤酒的有2000人，购买面包的有500人，
 * 同时购买尿布和啤酒的有800人，同时购买尿布和面包的有100人
 * 1. 支持度
 *    是指在所有项集中{X, Y} 出现的可能性。即项集中同时含有 X 和 Y 的概率： 该指标作为建立强关联规则的第一个门槛，
 *    衡量了所考察关联规则在“量”上的多少。通过设定最小阈值，剔除“出镜率”较低的无意义规则，保留出现较为频繁的项集所隐含的规则。
 *    如设定最小阈值为 5%，由于{尿布，啤酒}的支持度  800/10000=8%，满足最小阈值要求，成为频繁项集，保留规则，而{尿布，面包}
 *    的支持度为 100/10000 = 1%，则被剔除
 * 2. 置信度
 *    置信度表示在先决条件 X 发生的条件一，关联结果 Y 发生的概率。 这是生成强关联规则的第二个门槛。衡量了所考察的关联规则在 “质”上的可靠
 *    性。相似地，我们也需要对置信度设定一个最小阈值来进行筛选
 *    如，当设定置信度的最小阈值为 70%时，例如{尿布，啤酒}中，购买尿布时会购买啤酒的置信度为 800/1000=80%，保留。 崦购买
 *    啤酒时购买尿布的置信度为 800/2000=40% ， 则被剔除。 
 */
object Lesson01 {
  
  def main(args: Array[String]) : Unit = {
    val conf = new SparkConf
    conf.setMaster("local").setAppName("Lesson01")
    val sc = new SparkContext(conf);
    
    val lines = sc.textFile("D:\\bigdata\\spark\\SparkMLib\\01\\user_install_data\\000200_0")
    //val lines = sc.textFile("E:\\workspace\\scala\\spark\\sparkscala\\file\\lesson01.txt");
    val words = lines.map(line => {
      val tmp = line.split("\\s+")
      (tmp(0), tmp(1), tmp(2))
    })
    
    // 1 获取总数
    /*val cnt = lines.count()
    println("总数：" + cnt)
    val usrCnt = words.map(pair => (pair._2, 1)).reduceByKey(_+_).count()
    println("用户总数：" + usrCnt)
    val dateRdd = words.map(pair => (pair._1, 1)).reduceByKey(_+_)
    dateRdd.foreach(println)*/
    
    //指定任意连续的2天，计算每个用户第2天的新安装包名
    /*val dateRdd = words.map(pair => (pair._1, 1)).reduceByKey(_+_)
    val today = words.take(1)(0)._1 //取一个日期
    println("选取的时间： " + today)
    val todayRdd = words.filter(pair => pair._1 == today).map(pair => (pair._2, pair._3))
    
    val pattern = "yyyy-MM-dd"
    val sdf = new SimpleDateFormat(pattern);
    val date = sdf.parse(today)
    val next = new Date(date.getTime + 24 * 60 * 60 *1000)
    val nextStr = sdf.format(next)
    println("第二天： " + nextStr)
    
    val nextRdd = words.filter(pair => pair._1 == nextStr).map(pair => (pair._2, pair._3))
    val same = todayRdd.intersection(nextRdd)
    nextRdd.subtract(same).foreach(println)*/
    
    //指定任意1天的数据，根据每个用户的安装列表，统计每个包名的安装用户数量，由大到小排序，
    //并且取前1000个包名，最后计算这1000个包名之间的支持度和置信度
    val pickDate = "2016-04-08";
    //val dayRdd = words.filter(pair => pair._1 == pickDate);
    val dayRdd = words
    
    val pckUser = dayRdd.map(pair => (pair._3, 1)).reduceByKey(_+_)
    
    //安装排名前 1000 的应用
    val appCntRdd = pckUser.map(pair => (pair._2, pair._1))
        .sortByKey(false).map(pair => (pair._2, pair._1))
    println("排在前面的是：")
    val tmp = appCntRdd.map(pair => pair._1).take(10)
    tmp.foreach(println)
    val topApps = sc.broadcast(tmp)
    //topApps.foreach(println)
    
    val totalUser = dayRdd.map(pair => pair._2).distinct().count()
    println("总共有" + totalUser + "个用户")
    
    var userAppRdd = dayRdd.map(pair => (pair._2, pair._3)) // （用户，应用）
          .filter(pair => {
            topApps.value.contains(pair._2)
          })
          
    println(userAppRdd.count());
    val appPairRdd = userAppRdd.cartesian(userAppRdd)
        //过滤掉自己和自己的组合
        .filter(tuple=>tuple._1._1.equals(tuple._2._1) && !tuple._1._2.equals(tuple._2._2))
        //使得两个应用的顺序一致
        .map(tuple=>{
          if(tuple._1._2.compare(tuple._2._2) >= 0){
            (tuple._1._1, (tuple._1._2, tuple._2._2))
          }else{
            (tuple._1._1, (tuple._2._2, tuple._1._2))
          }
        }) // 得到 (用户, (应用1，应用2))
        .map(pair => (pair._2, 1)).reduceByKey(_+_) //计算同时安装两个应用的用户数
        .map(pair => (pair._1, pair._2/2)) // 第一组都会有2个相同的，所以总数要除以2
    println("==========================")
    appPairRdd.foreach(println)
    
    // 支持度
    val zhiChiDuRdd = appPairRdd.map(pair => (pair._1, pair._2.toFloat / totalUser))
    
    println("支持度：")
    zhiChiDuRdd.foreach(println)
    
    println("置信度：")
    // 应用组合中的第一个应用的置信度
    appPairRdd.map(pair=>(pair._1._1, (pair._1._2, pair._2)))
        .join(appCntRdd) // (应用1，((应用2，同时安装这两个应用的用户数), 应用1安装总数))
        .map(pair => (pair._1, pair._2._1._1, pair._2._1._2.toFloat/pair._2._2))
        .foreach(println)
    // 应用组合中的第二个应用的置信度  
    appPairRdd.map(pair=>(pair._1._2, (pair._1._1, pair._2)))
        .join(appCntRdd) // (应用2，((应用1，同时安装这两个应用的用户数), 应用2安装总数))
        .map(pair => (pair._1, pair._2._1._1, pair._2._1._2.toFloat/pair._2._2))
        .foreach(println)
  }
}